package com.crawler.store

import com.crawler.comm.CrawlerConfig
import com.crawler.service.PersistMessage
import com.crawler.utils.{CrawlerQueue, FileUtils}
import com.typesafe.config.ConfigFactory
import org.slf4j.LoggerFactory

import scala.collection.mutable.ListBuffer

/**
  * Created by tongtao.zhu on 2016/12/26.
  */
class Consumer extends Runnable {
  val log = LoggerFactory.getLogger(classOf[Consumer])
  val config = ConfigFactory.load()
  val crawlerConfig = new CrawlerConfig(config)
  override def run(): Unit = {
    try {
      while (true) {
        if (CrawlerQueue.link.size() > 0) {
          val messages = new ListBuffer[PersistMessage]
          while (!CrawlerQueue.link.isEmpty){
            messages.append(CrawlerQueue.link.pollLast())
          }
          messages.map(message => {
            val fileUtils = new FileUtils(s"${crawlerConfig.filePersist}/%s/%s".format(message.source, message.typ), crawlerConfig)
            fileUtils.crawlerDataStore(message)
          })
          messages.clear()
          Thread.sleep(2*60*1000)
        }
      }
    } catch {
      case e:Exception =>
    }
  }
}
